package com.example.rabbit.forth;

import com.example.rabbit.util.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * 发布订阅者模式
 * <p>
 * 1. 单个发布确认 ：同步
 * 2. 批量发布确认
 * 3. 异步发布确认
 * <p>
 * 对于 未确认消息重发 的解决办法？ ConcurrentLinkQueue
 *
 * @Author ifredomvip@gmail.com
 * @Date 2022/8/8 9:23
 * @Version 1.0.0
 * @Description
 **/
public class PublicConfirm {
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        // 单个确认
//        singleConfirm();
        // 批量确认
//        batchConfirm();
        // 异步批量确认
        asyncConfirm();
    }


    /**
     * 单个确认
     *
     * @throws Exception 异常
     */
    public static void singleConfirm() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();

        // 开启发布确认；
        channel.confirmSelect();

        channel.queueDeclare(queueName, false, false, false, null);

        // 开始时间
        Long startTime = System.currentTimeMillis();

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", queueName, null, message.getBytes());
            // 确认消息
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功" + i);
            }
        }

        // 结束时间
        Long endTime = System.currentTimeMillis();

        System.out.println("发布:" + MESSAGE_COUNT + " 个数，耗时 " + (endTime - startTime));
    }

    /**
     * 批量确认
     *
     * @throws Exception 异常
     */
    public static void batchConfirm() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();

        // 开启发布确认；
        channel.confirmSelect();

        channel.queueDeclare(queueName, false, false, false, null);

        // 开始时间
        Long startTime = System.currentTimeMillis();

        // 批量确认长度
        int batchSize = 100;

        // 批量发送消息  批量确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", queueName, null, message.getBytes());

            if (i % batchSize == 0) {
                // 确认消息
                boolean flag = channel.waitForConfirms();
                System.out.println("消息发送成功" + i);
            }
        }

        // 结束时间
        Long endTime = System.currentTimeMillis();

        System.out.println("批量发布:" + MESSAGE_COUNT + " 个数，耗时 " + (endTime - startTime));
    }

    /**
     * 异步确认
     *
     * @throws Exception 异常
     */
    public static void asyncConfirm() throws Exception {

        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        // 开启发布确认；
        channel.confirmSelect();


        /*
         * 线程安全有序的一个哈希表，书用于高并发的情况
         * 1. 轻松将序号与消息进行关联
         * 2. 轻松批量删除条目，只要给到序号
         * 3.支持高并发（多线程）
         *
         * 成功回调将 已发送的消息从跳表中删除
         * 失败回调将 未确认的消息重发
         *
         * 还有一点：失败回调中是拿不到消息内容，都失败了，自然不知道发送的消息内容是什么
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();


        // 开始时间
        Long startTime = System.currentTimeMillis();


        // 异步通知
        channel.addConfirmListener((sequenceNumber, multiple) -> {
            // code when message is confirmed
            System.out.println("异步确认成功" + sequenceNumber);

            // 2.删除已确认的消息
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber);
                confirmed.clear();
            } else {
                outstandingConfirms.remove(sequenceNumber);
            }

        }, (sequenceNumber, multiple) -> {
            // code when message is nack-ed
            System.out.println("异步确认失败" + sequenceNumber);
            // 未确认的消息
            String message = outstandingConfirms.get(sequenceNumber);
            System.err.format(
                    "Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",
                    message, sequenceNumber, multiple
            );
        });

        // 批量发送消息  批量确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;

            // 1.记录所有要发送的消息
            long sequenceNumber = channel.getNextPublishSeqNo();
            outstandingConfirms.put(sequenceNumber, message);

            channel.basicPublish("", queueName, null, message.getBytes());
        }

        // 结束时间
        Long endTime = System.currentTimeMillis();

        System.out.println("异步确认 发布:" + MESSAGE_COUNT + " 个数，耗时 " + (endTime - startTime));

    }
}

// 如果不使用lambada表达式，需要自己实现类
//class MyCallback implements ConfirmCallback {
//    private Long l;
//    private boolean b;
//
//    public MyCallback() {
//    }
//
//    public MyCallback(Long l, boolean b) {
//        this.l = l;
//        this.b = b;
//    }
//
//    @Override
//    public void handle(long l, boolean b) throws IOException {
//        System.out.println("异步确认成功");
//    }
//}